
//
// test load billions urls into store
//

#include <iostream>
#include <fstream>
#include <sstream>
#include <iomanip>
#include <string>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <boost/filesystem.hpp>
#include <boost/program_options.hpp>
namespace po = boost::program_options;


#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::posix_time;

#include "bocat/common/md5.hpp"
#include "bocat/common/url.hpp"
#include "bocat/common/logging.hpp"
#include "sakcab/store.hpp"

using namespace std;

#define	DEFAULT_TIMES	1024 * 256

#define	N_ROUND	100000

static bocat::common::Logger*   logger_p = NULL;

#define INFO(ARGS...)   logger_p->info(__LINE__,__FILE__,ARGS)
#define ERROR(ARGS...)  logger_p->error(__LINE__,__FILE__,ARGS)


float
getTimeDuration( ptime time_begin, ptime time_end )
{
	time_period	tp( time_begin, time_end );
	time_duration	td = tp.length();
	long	time_last_ms = td.total_milliseconds();
	float	time_last_s = time_last_ms / 1000.0f;

	return time_last_s;
}



void	test_load( const std::string& ini_filename, const std::string& ini_section, const std::string& file )
{
	sakcab::StoreManager	store_manager( ini_filename.c_str(), ini_section.c_str() );
	sakcab::Store*	store_p = store_manager.getStore();

	// test PUT
	{
		INFO( "testing LOAD ..." );
		ptime	time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;
		u_int32_t	count_round = 0;

		bocat::common::MD5	m;

		std::ifstream	ifs( file.c_str() );
		if ( !ifs ) {
			ERROR( "fail to open file %s", file.c_str() );
			return;
		}

		int32_t	value_i = 1;
		std::string	value( (char *)&value_i, sizeof(int32_t) );

		long	times = 0;
		long	count_bad_records = 0;
		long	bytes_total = 0;
		while ( !ifs.eof() ) {
			std::string	line;
			ifs >> line;

			boost::trim( line );
			if ( line.size() < 7 ) {
				continue;
			}

			std::string url_reversed;
			try  {
				bocat::common::URL  url( line );
				url_reversed = url.getReversedURL();
				boost::trim( url_reversed );
			}   
			catch (std::exception& e) {
				ERROR( "Exception : %s", e.what() );
				ERROR( "BAD LINE : %s", line.c_str() );
				//std::cerr << "Exception : " << e.what() << std::endl;
				//std::cerr << "BAD LINE : " << line << std::endl;
				++count_bad_records;
				continue;
			}   

			const std::string&	key = url_reversed;
			m.clear();
			m.update( key.c_str(), key.size() );
			string	kpin((const char* )(m.digest()), m.size());

			const std::string&	data = value;

			store_p->insert( kpin, key, data );

			bytes_total += data.size();

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
                INFO("  R-LOAD %u*%u records last %.3f seconds, tps=%.2f", ++count_round, N_ROUND, time_last_s, tps);

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
			}
		}

		ifs.close();

		ptime	time_stop = microsec_clock::local_time();
		float	time_last_s = getTimeDuration( time_start, time_stop );
		float	tps = times / time_last_s;
		float	bw = bytes_total / time_last_s / 1048576.0;
        INFO("    LOAD %u records last %.3f seconds, tps=%.2f, bw=%.2f MB/s", times, time_last_s, tps, bw);
	}

	// test SLIM
	{
		INFO( "testing SLIM ..." );
		ptime	time_start = microsec_clock::local_time();

		int rc = store_p->slim();
		if (rc < 0) {
			ERROR( "fail to slim() (rc=%d)", rc );
			return;
		}

		ptime	time_stop = microsec_clock::local_time();
		float	time_last_s = getTimeDuration( time_start, time_stop );
		INFO( "    SLIM %d records last %.2f seconds", rc, time_last_s );
	}


	/*
	// test ERASE
	{
		INFO( "testing ERASE ..." );
		ptime	time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;
		u_int32_t	count_round = 0;

		bocat::common::MD5	m;

		for (int i=0; i<times; i++) {
			ostringstream	os_key;      os_key << "KEY--" << setw(8) << setfill('0') << i;

			string	key = os_key.str();

			m.clear();
			m.update( key.c_str(), key.size() );
			string	kpin((const char* )(m.digest()), m.size());

			store_p->erase( kpin, key );

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
                INFO("  R-ERASE %u*%u records last %.3f seconds, tps=%.2f", ++count_round, N_ROUND, time_last_s, tps);

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
			}
		}

		ptime	time_stop = microsec_clock::local_time();
		float	time_last_s = getTimeDuration( time_start, time_stop );
		float	tps = times / time_last_s;
        INFO("  ERASE %u records last %.3f seconds, tps=%.2f", times, time_last_s, tps);
	}
	*/
}


int	main(int argc, char* argv[])
{
	std::string	ini_filename;
	std::string	ini_section;
	//int 	times = 0;
	//int 	data_size = 0;
	bool	verbose = false;
	std::string	file;

    std::string log_config;
    std::string log_section;

    try {

        po::options_description desc("Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("verbose,v", "output verbose message")
			//("times", po::value<int>(&times)->default_value(DEFAULT_TIMES), "number of test to execute")
			//("data-size", po::value<int>(&data_size)->default_value( 8192 ), "size of data chunk")
            ("ini-file", po::value<std::string>(&ini_filename)->default_value("store.ini"), "INI configure filename")
            ("ini-section", po::value<std::string>(&ini_section)->default_value("sakcab::store"), "INI configure section")

            ("load", po::value<std::string>(&file), "data file to load")

            ("log-config", po::value<std::string>(&log_config), "path to logging config file")
            ("log-section", po::value<std::string>(&log_section), "section name of logging config")
        ;   

        po::variables_map vm;    
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);    

        if (vm.count("help")) {
            std::cout << desc << std::endl;
            return 1;
        }   

		if (vm.count("verbose")) {
			verbose = true;
		}

		if (vm.count("load") < 1 ) {
			std::cout << "No data file to load\n\n";
			std::cout << desc << std::endl;
		}
    }   
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }   
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
		return 2;
    }   

    logger_p = bocat::common::getLogger();
    if ( log_config.size() > 0 ) {
        logger_p->configure( log_config.c_str(), log_section.c_str() );
    }
	
	test_load( ini_filename, ini_section, file );


	return	0;
}


